DXP-2486 Add streamx run, stream and batch commands for StreamX 2.0 (and remove streamx publish/unpublish/cloud/dev/init commands for now) #219

stan-r-ds · 2025-12-29T13:34:55Z

📋 Type of the Changes

Breaking change
Non-breaking change
Bug fix / minor change

🛠 Changes being made

Added streamx commands for StreamX 2.0 (and removed the original run/stream/batch):

run_v2
stream_v2
batch_v2

Removed streamx commands:

publish and unpublish: permanently, as discussed
cloud, dev and init: removed for now, since adjusting them to StreamX 2.0 was not part of the task, but need the whole codebase to compile and pass tests

✅ Checklist

My code follows the code standards of this project
Changed code is covered with unit tests
I have updated READMEs and java docs (if applicable)

…and remove streamx publish/unpublish/cloud/dev/init commands for now)

stan-r-ds · 2025-12-29T14:06:45Z

.github/workflows/ci-test-build.yaml

        uses: actions/setup-java@v3
        with:
-          java-version: '17'
+          java-version: '21'


The builds fail on java 17 when importing com.streamx ingestion client that's built with java 21

stan-r-ds · 2025-12-29T14:08:30Z

...ain/java/dev/streamx/cli/command/ingestion/batch/resolver/substitutor/StringSubstitutor.java

  @Override
  public Map<String, String> createSubstitutionVariables(String payloadPath,
-      String channel, String relativePath) {
+      String eventType, String eventSource, String relativePath) {


In the eventsource.yaml files:

channel is now replaced with eventType

and eventSource is added to them, as a mandatory field in CloudEvents API

stan-r-ds · 2025-12-29T14:11:26Z

core/src/main/java/dev/streamx/cli/command/ingestion/batch/BatchCommand.java

-  private final Map<String, Publisher<JsonNode>> publishersCache = new HashMap<>();
-  private final Map<String, String> schemaTypesCache = new HashMap<>();
+  // TODO "_v2" is a temporary postfix for now
+  public static final String COMMAND_NAME = "batch_v2";


Added those postfixes instead of adding an additional experimental word in the command, to not break the current limitations that each command has a single word name

stan-r-ds · 2025-12-29T14:13:36Z

core/src/main/java/dev/streamx/cli/command/ingestion/batch/BatchCommand.java

-  @Inject
-  IngestionMessageJsonFactory ingestionMessageJsonFactory;
-
-  private State state;


Now a method parameter instead of a mutable field

stan-r-ds · 2025-12-29T14:13:52Z

core/src/main/java/dev/streamx/cli/command/ingestion/batch/BatchCommand.java

-          if (publisher == null) {
-            publisher = client.newPublisher(getChannel(), JsonNode.class);
-            publishersCache.put(getChannel(), publisher);
-          }


Now a single publisher

stan-r-ds · 2025-12-29T15:09:16Z

e2e-tests/src/test/java/dev/streamx/cli/StreamxCliPublicationIT.java

  @ConfigProperty(name = "streamx.cli.e2e.web.delivery.url", defaultValue = "http://localhost:8087/")
  String webDeliveryPortUrl;
-  @ConfigProperty(name = "streamx.cli.e2e.nginx.url", defaultValue = "http://localhost:8089/overridden/")
-  String nginxPortUrl;


To make things simplier maybe we don't need to test both web sink and nginx in CLI - this is rather the responsibility of streamx-service-mesh/runner (and its tested there)

stan-r-ds · 2025-12-29T15:12:34Z

e2e-tests/src/test/java/dev/streamx/cli/StreamxCliPublicationIT.java

-              CLI_SHORT_TIMEOUT_IN_SEC)
-    );
+    String url = webDeliveryPortUrl + resourcePath;
+    httpValidator.validate(url, expectedStatusCode, expectedBody, CLI_TIMEOUT);


Testing just in the Web Sink is enough 💭

stan-r-ds · 2025-12-29T15:13:13Z

e2e-tests/src/test/resources/batch/unpublish/page/index.html

@@ -0,0 +1 @@
+<h1>Hello World!</h1>


...could have empty content, since it's for unpublishing - but still the file must exist to be picked up

stan-r-ds · 2025-12-29T15:14:03Z

pom.xml

-    <streamx.version>1.0.16</streamx.version>
-    <ingestion-client.version>1.1.1</ingestion-client.version>
-    <streamx-operator.version>0.0.16</streamx-operator.version>
+    <streamx.version>2.0.6</streamx.version>


Merge client and runner versions to single release of streamx-service-mesh

stan-r-ds · 2025-12-29T15:14:20Z

pom.xml

+        <executions>
+          <execution>
+            <goals>
+              <goal>integration-test</goal>


Enable running the e2e tests

stan-r-ds · 2025-12-29T15:59:57Z

.../src/main/java/dev/streamx/cli/command/ingestion/batch/walker/EventSourceFileTreeWalker.java

 import dev.streamx.cli.command.ingestion.batch.EventSourceDescriptor;
 import dev.streamx.cli.command.ingestion.batch.exception.EventSourceDescriptorException;
 import dev.streamx.cli.util.FileUtils;
-import io.quarkus.logging.Log;


This class makes jacoco coverage measurer lost - makes the whole class have 0% coverage, which may make it look unused

This also fixes the
[WARNING] Rule violated for bundle streamx-cli-core: instructions covered ratio is 0.65, but expected minimum is 0.70

stan-r-ds · 2025-12-29T16:00:41Z

.../src/main/java/dev/streamx/cli/command/ingestion/batch/walker/EventSourceFileTreeWalker.java


 public class EventSourceFileTreeWalker extends SimpleFileVisitor<Path> {

+  private final Logger logger = Logger.getLogger(EventSourceFileTreeWalker.class);


Other classes in this project also do this

stan-r-ds · 2025-12-30T10:53:59Z

...est/java/dev/streamx/cli/command/meshprocessing/MeshDefinitionResolverInterpolationTest.java


  @Inject
  MeshDefinitionResolver uut;



Added test for interpolating with default value, such as:

ref: "${config.source.interpolated:inbox.pages}"

It works as expected 👍

stan-r-ds · 2025-12-30T10:54:18Z

...est/java/dev/streamx/cli/command/meshprocessing/MeshDefinitionResolverInterpolationTest.java

-    System.setProperty("config.image.interpolated", "value");
+  void shouldResolveWithMandatoryAndOptionalPropertiesDefined() throws IOException {
+    System.setProperty("config.image.interpolated", "image-1");
+    System.setProperty("config.source.interpolated", "source-1");


Default value is overridden here

stan-r-ds · 2025-12-30T10:55:44Z

.github/workflows/ci-test-build.yaml

      - name: Build project
        run: |
-          ./mvnw clean verify -P all-tests
+          ./mvnw clean verify -P all-tests -Dnative


leverage the multi-arch feature to allow running mesh (in integration tests) also on both linux and mac host systems

stan-r-ds · 2025-12-31T09:38:58Z

core/src/main/java/dev/streamx/cli/BannerPrinter.java


  private static final Set<String> COMMANDS_REQUIRING_PRINTING_BANNER =
-      Set.of(DevCommand.COMMAND_NAME, RunCommand.COMMAND_NAME);
+      Set.of(RunCommand.COMMAND_NAME);


The not migrated (so - removed) commands are

dev

cloud (deploy/undeploy)

init

publish

unpublish

stan-r-ds · 2025-12-31T09:40:12Z

core/src/main/java/dev/streamx/cli/command/ingestion/BaseIngestionCommand.java

-      doRun(client);
-    } catch (UnsupportedChannelException e) {
-      throw new ParameterException(spec.commandLine(), e.getMessage());
-    } catch (StreamxClientConnectionException e) {


Those exception types are gone

stan-r-ds · 2025-12-31T09:40:30Z

core/src/main/java/dev/streamx/cli/command/ingestion/BaseIngestionCommand.java

-    } catch (StreamxClientConnectionException e) {
-      throw new UnableToConnectIngestionServiceException(ingestionClientConfig.url(), e);
+      Publisher publisher = client.newPublisher();
+      perform(publisher);


Inlined the doRun method here

stan-r-ds · 2025-12-31T12:10:55Z

core/src/main/java/dev/streamx/cli/command/ingestion/batch/BatchCommand.java

+        state.eventSubject,
+        state.eventType,
+        state.eventSource,
+        state.payload // TODO rename payload to data everywhere


I think worth doing this in a separate PR, since a lot of files would get changed, making the PR too hard to read. Internally "payload" is assigned as cloud event's "data" ✅

stan-r-ds · 2025-12-31T12:12:27Z

core/src/main/java/dev/streamx/cli/command/ingestion/batch/CloudEventBuilder.java

+        .withDataContentType("application/json")
+        .withData(JsonCloudEventData.wrap(data))
+        .build();
+  }


Needed only by the batch command to create the event from event descriptor and data payload files

stan-r-ds · 2025-12-31T12:33:28Z

core/src/test/resources/dev/streamx/cli/command/ingestion/stream/valid-json.stream

+  "subject" : "homepage.html",
+  "time" : "2025-12-23T10:28:23.435253Z",
+  "data" : {
+    "content" : "<h1>This is Home Page!</h1>",


This works because next to this file we have .stream.properties file with content:
json.paths.to.encode.to.base64=/data/content

stan-r-ds · 2025-12-31T12:35:26Z

core/src/test/resources/filtered/mesh.yaml

+          events: {}
+        outgoing:
+          relayed-events: {}
+


This mesh.yaml is used only by RunCommandTest, so it must be valid and "runnable"

stan-r-ds · 2025-12-31T12:38:08Z

core/src/test/resources/filtered/mesh-interpolated.yaml

-defaultImageTag: ${streamx.version}-jvm
+descriptors:
+  relay:
+    type: "processing"


used only in interpolation unit test

stan-r-ds · 2025-12-31T12:38:27Z

core/src/test/resources/filtered/mesh-interpolated.yaml

  cli:
    outgoing:
-      - "pages"
+      - ref: "${config.source.interpolated:inbox.pages}"


Added test for property with default value.
Turned out it is applied

stan-r-ds · 2025-12-31T12:39:46Z

e2e-tests/pom.xml

+      <groupId>org.assertj</groupId>
+      <artifactId>assertj-core</artifactId>
+      <scope>compile</scope>
+    </dependency>


This module is a bit weird:

it has tests in src/test

but it has test utils in src/main 🤷

That's why we can use assertj. It gives way better error messages while troubleshooting "java -jar runner.jar" commands

stan-r-ds · 2025-12-31T12:40:20Z

e2e-tests/pom.xml

        <directory>src/test/resources/filtered</directory>
        <includes>
          <include>mesh.yaml</include>
-          <include>overridden-nginx-conf</include>


Maybe no need to test this feature in the CLI, it's tested at source projects

stan-r-ds · 2025-12-31T12:41:01Z

e2e-tests/src/main/java/dev/streamx/cli/test/tools/validators/HttpValidator.java

+      String responseBody = EntityUtils.toString(httpEntity);
+      assertThat(responseBody).describedAs(url).contains(expectedBody);
+    });
  }


Added an overload to test also binary expected body (to verify an image will be published to StreamX with unchanged content)

stan-r-ds · 2025-12-31T12:41:48Z

e2e-tests/src/main/java/dev/streamx/cli/test/tools/validators/HttpValidator.java

-      String responseBody = EntityUtils.toString(response.getEntity());
-      logger.info("Request to " + url
-          + " return statusCode " + actualStatusCode
-          + " and body " + responseBody);


Thanks to assertj .describedAs() as well as its descriptive error messages - we don't need to log manually anymore

stan-r-ds · 2025-12-31T12:42:57Z

e2e-tests/src/main/java/dev/streamx/cli/test/tools/validators/ProcessOutputValidator.java

+                  .describedAs(() -> "Full output is:\n"
+                                     + String.join("\n", process.getCurrentOutputLines())
+                                     + "\n"
+                                     + String.join("\n", process.getCurrentErrorLines()))


Adding very helpful logs.

Until now, the tests tested that "java -jar"'s process success output logs contain expected text.

Now - when they don't - the full process output (including errors) is logged

stan-r-ds · 2025-12-31T12:44:32Z

e2e-tests/src/test/java/dev/streamx/cli/StreamxCliPublicationIT.java

  }

-  private void runStreamxCommand(String command, String expectedOutput, long timeoutInS) {
+  private void runStreamxCommand(String command, String expectedOutput, Duration timeout) {


No need to specify the timeout unit in parameter name, if we can use Duration object

stan-r-ds · 2025-12-31T12:45:17Z

e2e-tests/src/test/java/dev/streamx/cli/StreamxCliPublicationIT.java

+  }

-  @ParameterizedTest
-  @MethodSource("testCases")


Those were test cases for publish command, and it was decided that

we remove publish and unpublish

because no one uses them

and they handle just a single ingestion while the jar start time is long

we don't want to manage the wide variety of options described in this method

Instead, the test class was modified to verify end-to-end:

batch publish and unpublish

stream publish and unpublish

stan-r-ds · 2025-12-31T12:46:50Z

e2e-tests/src/test/java/dev/streamx/cli/StreamxCliPublicationIT.java

    );
+
+    validateStreamxPage("ds.png",
+        Files.readAllBytes(Path.of("src/test/resources/batch/publish/asset/ds.png")));


The image is published to real StreamX and here we verify that when the image is downloaded via http from web sink service url - the bytes are unchanged

stan-r-ds · 2026-01-05T09:25:29Z

e2e-tests/src/main/java/dev/streamx/cli/StreamxTerminalCommandProducer.java

+        .map(File::toPath)
+        .map(Path::toAbsolutePath)
+        .map(Path::normalize)
+        .map(Path::toString)


Remove any "/../" from the logged path

stan-r-ds · 2026-01-05T09:25:45Z

e2e-tests/src/main/java/dev/streamx/cli/test/tools/terminal/OsTerminalCommandRunner.java

      processes.add(shellProcess);
      return shellProcess;
    } catch (IOException e) {
+      logger.info("Error running terminal command: " + command);


Deduplicate log messages for success

DXP-2486 Add streamx run, stream and batch commands for StreamX 2.0 (…

243064b

…and remove streamx publish/unpublish/cloud/dev/init commands for now)

stan-r-ds requested a review from a team as a code owner December 29, 2025 13:34

stan-r-ds added 5 commits December 29, 2025 14:43

mockito

eb63811

java 21

288ccf8

add missing validation

e8ac81a

Remove Resource class

9761683

Restore banner

46a6874

stan-r-ds commented Dec 29, 2025

View reviewed changes

stan-r-ds added 3 commits December 29, 2025 16:25

Remove unnecessary plugin

74f6c6d

Use cli source

cad43e5

Increase test coverage

c8ca778

stan-r-ds commented Dec 29, 2025

View reviewed changes

stan-r-ds added 5 commits December 29, 2025 17:04

Use jvm

9776c77

Use temurin

dca7a18

Move test file, add test for default values

18fdda4

Better newline diff comparison

f08a25d

Use native

16501c8

stan-r-ds force-pushed the DXP-2486-add-run-stream-and-batch-commands-for-streamx-2.0 branch from 2942184 to 16501c8 Compare December 30, 2025 09:26

restore {}

dd8a24b

stan-r-ds commented Dec 30, 2025

View reviewed changes

Improve handling of ByteBuffer fields

0b9fcd3

stan-r-ds commented Dec 31, 2025

View reviewed changes

CloudEventBuilder fixes

2145fc8

stan-r-ds commented Dec 31, 2025

View reviewed changes

fixes

403fd23

stan-r-ds commented Dec 31, 2025

View reviewed changes

stan-r-ds added 5 commits December 31, 2025 14:17

fix codestyle

d3f7eae

Small fix

1f13cc6

Use new streamx-service-mesh version

811cab5

Don't use native

6b6e62d

Minor fixes

ec213a6

stan-r-ds commented Jan 5, 2026

View reviewed changes

stan-r-ds added 4 commits January 5, 2026 10:45

add debug logging

c6eba11

Remove repository volume

cd8b684

Volumes cleanup

976a3d6

Temporarily allow releasing preview on demand

0158222

stan-r-ds force-pushed the DXP-2486-add-run-stream-and-batch-commands-for-streamx-2.0 branch from 49665df to 0158222 Compare January 5, 2026 10:30

stan-r-ds added 5 commits January 5, 2026 12:03

Temporarily embed releasing preview on demand into PR CI

58337de

Make release preview to use specific branch

8e1100e

Restore recent changes regarding preview releasing

3ba20b4

Rename json.paths.to.encode.to.base64 to json.fields.as-base64

ee83d91

Formatting agnostic verification

a0b9cc6


		public class EventSourceFileTreeWalker extends SimpleFileVisitor<Path> {

		private final Logger logger = Logger.getLogger(EventSourceFileTreeWalker.class);

DXP-2486 Add streamx run, stream and batch commands for StreamX 2.0 (and remove streamx publish/unpublish/cloud/dev/init commands for now) #219

Are you sure you want to change the base?

DXP-2486 Add streamx run, stream and batch commands for StreamX 2.0 (and remove streamx publish/unpublish/cloud/dev/init commands for now) #219

Uh oh!

Conversation

stan-r-ds commented Dec 29, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

📋 Type of the Changes

🛠 Changes being made

✅ Checklist

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stan-r-ds Dec 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stan-r-ds Dec 31, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stan-r-ds Dec 31, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stan-r-ds Dec 31, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stan-r-ds Dec 31, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

stan-r-ds commented Dec 29, 2025 •

edited

Loading

stan-r-ds Dec 30, 2025 •

edited

Loading

stan-r-ds Dec 31, 2025 •

edited

Loading

stan-r-ds Dec 31, 2025 •

edited

Loading

stan-r-ds Dec 31, 2025 •

edited

Loading

stan-r-ds Dec 31, 2025 •

edited

Loading

stan-r-ds Dec 31, 2025 •

edited

Loading